package io.gatling.extensions.kafka.builder

import java.util

import com.typesafe.scalalogging.StrictLogging
import io.gatling.core.action.Action
import io.gatling.core.action.builder.ActionBuilder
import io.gatling.core.structure.ScenarioContext
import io.gatling.extensions.kafka.action.KafkaConsumerAction
import io.gatling.extensions.kafka.protocol.KafkaConsumerProtocol

case class KafkaConsumerActionBuilder(name: String) extends ActionBuilder with StrictLogging {

  val topics: util.ArrayList[String] = new util.ArrayList[String]()

  def topic(topics: util.ArrayList[String]): KafkaConsumerActionBuilder = {
    this.topics.addAll(topics)
    this
  }

  override def build(ctx: ScenarioContext, next: Action): Action = {
    import ctx._
    KafkaConsumerAction(
      name,
      topics,
      protocolComponentsRegistry.components(KafkaConsumerProtocol.kafkaConsumerKey),
      coreComponents,
      throttled,
      next
    )
  }
}
